package com.pingan.haofang.searchcloud.common.amqp;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.pingan.haofang.searchcloud.common.utils.RequestTraceUtil;
import com.rabbitmq.client.Channel;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * 消息侦听模板
 * 
 * @author LUYI374
 * @date 2017年3月7日
 * @since 1.0.0
 */
public abstract class BaseMessageListener {

    static Logger LOG = LoggerFactory.getLogger(BaseMessageListener.class);

    public static final int ACK_RETRY_COUNT = 3;

    Gson GSON = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
    /**
     * 处理消息
     * 
     * @param clazz
     * @param datas
     * @param queueName
     * @param channel
     * @param deliveryTag
     * @param encoding
     * @param retryCount
     * @param messageHandler
     */
    protected <T> void handleMessage(Class<T> clazz, byte[] datas, String queueName, Channel channel, long deliveryTag,
            String encoding, int retryCount, MessageHandler<T> messageHandler) {
        try {
            RequestTraceUtil.requestTraceStart();

            String messageBody = new String(datas, encoding);
            if (StringUtils.isNotBlank(queueName)) {
                LOG.info("message received, queue: {}, body: {}", queueName, messageBody);
            } else {
                LOG.info("message received, body: {}", messageBody);
            }

            // 解析消息体
            final T message = parseMessage(messageBody, channel, deliveryTag, clazz);

            // 如果消息体解析失败，则直接丢弃消息
            if (message == null) {
                nack(channel, deliveryTag, false);
                return;
            }

            messageHandler.setChannel(channel);
            messageHandler.setDeliveryTag(deliveryTag);
            messageHandler.setRetryCount(retryCount);

            messageHandler.handle(message);
        } catch (Throwable e) {
            LOG.error(e.getMessage(), e);
            nack(channel, deliveryTag, true);
        } finally {
            RequestTraceUtil.requestTraceEnd();
        }
    }

    /**
     * 解析消息体
     * 
     * @param messageBody
     * @param channel
     * @param deliveryTag
     * @param clazz
     * @return
     */
    private <T> T parseMessage(String messageBody, Channel channel, long deliveryTag, Class<T> clazz) {
        T message = null;
        try {
            message =  GSON.fromJson(messageBody, clazz);
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
        return message;
    }

    /**
     * 消息确认
     * 
     * @param channel
     * @param deliveryTag
     */
    private void ack(Channel channel, long deliveryTag) {
        int current = ACK_RETRY_COUNT;
        while (current > 0) {
            try {
                channel.basicAck(deliveryTag, false);
                break;
            } catch (IOException e) {
                LOG.error(e.getMessage(), e);
                current--;
            }
        }
    }

    /**
     * 消息拒绝
     * 
     * @param channel
     * @param deliveryTag
     * @param retry 是否重试
     */
    private void nack(Channel channel, long deliveryTag, boolean retry) {
        int current = ACK_RETRY_COUNT;
        while (current > 0) {
            try {
                channel.basicNack(deliveryTag, false, retry);
                break;
            } catch (IOException e) {
                LOG.error(e.getMessage(), e);
                current--;
            }
        }
    }

    /**
     * 消息处理器
     * 
     * @author LUYI374
     * @date 2017年3月7日
     * @since 1.0.0
     */
    public abstract class MessageHandler<T> {

        private Channel channel;
        private long deliveryTag;
        private int retryCount;

        /**
         * 处理消息
         * 
         * @param message
         */
        public void handle(T message) {

            // 校验消息体合法性，如果不合法，直接丢弃消息
            if (!doValidate(message)) {
                LOG.error("message validate failed, ignore it");
                nack(channel, deliveryTag, false);
                return;
            }

            int current = retryCount;
            while (current > 0) {
                try {
                    // 入库
                    doProcess(message);

                    // 入库完成发送ACK
                    ack(channel, deliveryTag);
                    break;
                } catch (Exception e) {
                    LOG.error(e.getMessage(), e);
                    current--;

                    // 如果超过重试次数，则丢弃该消息
                    if (current <= 0) {
                        LOG.error(e.getMessage(), e);
                        nack(channel, deliveryTag, false);
                    }
                }
            }
        }

        /**
         * 消息体校验
         * 
         * @param message
         */
        public boolean doValidate(T message) {
            return true;
        }

        /**
         * 消息处理
         * 
         * @param message
         */
        public abstract void doProcess(T message);

        public void setChannel(Channel channel) {
            this.channel = channel;
        }

        public void setDeliveryTag(long deliveryTag) {
            this.deliveryTag = deliveryTag;
        }

        public void setRetryCount(int retryCount) {
            this.retryCount = retryCount;
        }

    }
}
